/***
* ==++==
*
* Copyright (c) Microsoft Corporation.  All rights reserved.
*
* ==--==
* =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
*
* actor_process.h
*
* C++ Actors: actor, the main user-facing C++ Actors APIs.
*
* =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
****/

#pragma once

#include "ppl.h"
#include "pplxtasks.h"

#include "actor_public.h"
#include "context.h"
#include "pid.h"
#include "procdir.h"
#include "procstore.h"

namespace actors
{

    /// <summary>
    /// actor
    ///
    /// The base class of all actors defined by users of the C++ Actors library
    /// </summary>
    class actor : public Concurrency::agent, public _IMessageDispatcher, protected pplx::scheduler
    {
    public:
        /// <summary>
        /// Default constructor
        /// </summary>
        _DISPIMP actor();

        /// <summary>
        /// Destructor
        /// </summary>
        _DISPIMP virtual ~actor();

        /// <summary>
        /// Explicitly register an actor instance that was created through other 
        /// means than runtime hosting (such as started using Agent::Start()).
        /// </summary>
        _DISPIMP void _register_instance();

        /// <summary>
        /// Get a pointer to the current instance, useful when it is necessary to 
        /// have access to an actor from a library function, for example.
        /// </summary>
        _DISPIMP static actor *current();
        _DISPIMP static void set_current(actor *);

        //
        // Actor Storage and Directory Implementation Definition
        //

        /// <summary>
        /// Define the actor checkpoint storage implementation object.
        /// </summary>
        /// <param name="store">A pointer to a ProcessStorage implementation</param>
        _DISPIMP static void set_proc_storage(ProcessStorage *store);

        /// <summary>
        /// Define the actor checkpoint storage implementation object.
        /// </summary>
        /// <param name="directory">A pointer to a ProcessDirectory implementation</param>
        _DISPIMP static void set_proc_dir(ProcessDirectory *directory);

        //
        // Actor Directory Access
        //

        /// <summary>
        /// Enter a record of a factory in the system's actor directory.
        /// </summary>
        /// <param name="type">The type info for the actor type</param>
        /// <param name="address">The address at which the factory is now available</param>
        /// <returns>True if a directory implementation was found and a record could be added.</returns>
        _DISPIMP static bool add_to_directory(const type_info &type, const std::string &address);

        /// <summary>
        /// Template version of the previous function.
        /// </summary>
        template<typename T>
        static bool add_to_directory(const std::string &address) { return add_to_directory(typeid(T), address); }

        /// <summary>
        /// Remove any records of a factory in the system's actor directory.
        /// </summary>
        /// <param name="type">The type info for the actor type</param>
        /// <param name="address">The address at which the factory is no longer available</param>
        /// <returns>True if a directory implementation was found and the record could be removed.</returns>
        _DISPIMP static bool remove_from_directory(const type_info &type, const std::string &address);

        /// <summary>
        /// Template version of the previous function.
        /// </summary>
        template<typename T>
        static bool remove_from_directory(const std::string &address) { return remove_from_directory(typeid(T), address); }

        /// <summary>
        /// find all the available factories of a given actor type.
        /// </summary>
        /// <param name="type">The type info for the actor type</param>
        /// <param name="locations">Out parameter: a set of addresses where the actor type is available</param>
        /// <returns>True if a directory implementation was found. The set of locations may be empty even
        /// if the function returns true.</returns>
        _DISPIMP static bool find(const type_info &type, std::vector<std::string> &locations);

        /// <summary>
        /// Template version of the previous function.
        /// </summary>
        template<typename T>
        static bool find(std::vector<std::string> &locations) { return find(typeid(T), locations); }

        /// <summary>
        /// Determine the actor type available at a given address, if any.
        /// </summary>
        /// <param name="address">The address to query</param>
        /// <param name="type">Out parameter: the type info found at the given address</param>
        /// <returns>True if a directory implementation and a factory were both found, i.e. 'type' could be filled in.</returns>
        _DISPIMP static bool find(const std::string &address, type_info &type);

    #pragma region "spawn()"
        /// <summary>
        /// Spawn is used to create a local or remote actor and return an
        /// identifier that may be used to communicate with the process.
        /// </summary>
        /// <typeparam name="ActorType">The actor type, derived from 'actor'</typeparam>
        /// <param name="uri">
        /// The address of the host and factory where the actor should be created.
        /// If omitted, the actor will be created in the local environment.</param>
        template<typename ActorType>
        static PID spawn(const std::string &uri_string = "")
        {
            http::uri uri(uri_string);
            
            if (!uri.empty() && uri.is_authority())
            {
                return PID().invalidate();
            }

            tcp_listener *listener = tcp_listener::find(uri);

            if ( listener != nullptr )
            {
                actor *proc = reinterpret_cast<actor *>(new ActorType);
                actor_info p_info = proc->self_pid->m_actor_info;
                if ( !p_info ) return PID();

                p_info->process = proc;

                tcp_listener::_register_actor(p_info->guid, false, p_info);

                process_factory factory = listener->_get_factory(typeid(ActorType).name());
                if ( factory ) proc->set_context(listener, factory->Context);

                proc->initialize(ACTR_Create);
                proc->start();
            
                return PID(p_info);
            }
            else
            {
                return PID(_actor_info::_spawn_remote(typeid(ActorType).name(), uri));
            }
        }

        /// <summary>
        /// Spawn is used to create a local or remote actor and return an
        /// identifier that may be used to communicate with the actor.
        /// </summary>
        /// <param name="uri">
        /// The address of the host and factory where the actor should be created.
        /// If omitted, the actor will be created in the local environment.</param>
        _DISPIMP static PID spawn(const std::string &uri);
        _DISPIMP static PID spawn(const std::string &baseUri, const std::string &resource);

        /// <summary>
        /// Respawn is used to restart an actor after it has crashed.
        /// </summary>
        /// <param name="data">A buffer containing the state passed from the actor with the fault, if any</param>
        _DISPIMP static PID respawn(fault_msg msg);

        /// <summary>
        /// Spawn a group of actors of the same type, given a count and a host address. All
        /// spawned actors will be executing in the same host.
        /// </summary>
        /// <param name="uri">The address at which to create the actors</param>
        _DISPIMP static PID::Group spawn_group(size_t count, const std::string &uri);
        _DISPIMP static PID::Group spawn_group(size_t count, const std::string &baseUri, const std::string &resource);

        /// <summary>
        /// Spawn a group of actors of the same type, given a count and a host address. All
        /// spawned actors will be executing in the same host.
        /// </summary>
        /// <typeparam name="ActorType">The actor type, derived from 'actor'</typeparam>
        /// <param name="uri">
        /// The address at which to create the actors. If omitted, actors will be create in the
        /// local environment.
        /// </param>
        template<typename ActorType>
        static PID::Group spawn_group(size_t count, const std::string &uri = "")
        {
            PID::Group result;

            if ( tcp_listener::find(uri) )
            {
                for (int i = 0; i < count; i++)
                {
                    result.push_back(spawn<ActorType>());
                }
                return result;
            }
            else
            {
                for (int i = 0; i < count; i++)
                {
                    result.push_back(spawn<ActorType>(uri));
                }
                return result;
            }
        }

        /// <summary>
        /// Spawn a group of actors of the same type, given a count and
        /// a group of hosts. The actors will be distributed in round-robin fashion
        /// among the hosts, meaning that no host runs more than one actor more than
        ////any other host.
        /// </summary>
        /// <param name="hosts">The base addresses at which to create the actors.</param>
        /// <param name="resource">The relative addresses at which to create the actors.</param>
        _DISPIMP static PID::Group spawn_group(size_t count, const std::vector<std::string> &hosts, const std::string &resource);

        /// <summary>
        /// Spawn a group of actors of the same type, given a count and 
        /// a group of hosts. The actors will be distributed in round-robin fashion
        /// among the hosts, meaning that no host runs more than one actor more than
        ////any other host.
        /// </summary>
        /// <typeparam name="ActorType">The actor type, derived from 'actor'</typeparam>
        /// <param name="hosts">The base addresses at which to create the actors.</param>
        /// <param name="resource">The relative addresses at which to create the actors.</param>
        template<typename ActorType>
        static PID::Group spawn_group(size_t count, const std::vector<std::string> &hosts)
        {
            PID::Group result;

            //
            // Divide the number of spawned instances equally between the
            // members of the group.
            //
            int workers = (int) hosts.size();
            int numPerRank = (int) ((count % workers == 0) ? (count/workers) : (count/workers)+1);

            for (auto rank = hosts.begin(); rank != hosts.end() && count > 0; rank++, count -= numPerRank)
            {
                int cnt = (int) ((count < 2*numPerRank) ? count : numPerRank);
                auto grp = spawn_group<ActorType>(cnt, *rank);
                result.insert(result.end(), grp.begin(), grp.end());
            }
            return result;
        }
    #pragma endregion

    protected:

    #pragma region "receive()"
        /// <summary>
        /// While 'accept' receives data from the single inbox of a given process, 'receive' gets data from any ISource.
        /// This allows the programmer to use all the Dev10 agents library construct within his or her actor code.
        ///
        /// Using the version of receive that is available in the 'Concurrency' namespace will likely result in deadlocks,
        /// since it doesn't know about the per-process lock or that it is necessary to give it up while waiting.
        /// </summary>
        template<typename Payload, typename Functor>
        bool receive(Concurrency::ISource<Payload> *src, Functor &action)
        {
            if ( src == nullptr ) throw std::exception("Invalid argument to 'receive': src cannot be null");

            void **_holder = new void *[2];

            Concurrency::single_assignment<Payload> *sa = new Concurrency::single_assignment<Payload>();

            auto my_pid = self_pid;

            Concurrency::call<Payload> *call = new Concurrency::call<Payload>(
                [my_pid,this,_holder,src,action](Payload data)
                {
                    Concurrency::call<Payload> *_call = reinterpret_cast<Concurrency::call<Payload> *>(_holder[0]);
                    Concurrency::single_assignment<Payload> *_sa = reinterpret_cast<Concurrency::single_assignment<Payload> *>(_holder[1]);

                    auto _this = this;
                    auto holder = _holder;
                    auto _action = action;
                    auto my_pid = self_pid;

                    src->unlink_target(_sa);
                    _sa->unlink_target(_call);

                    data_lock.lock();

                    _schedule(CreateContinuation(
                        [my_pid,_this,holder,_action,_sa,_call,data]()
                        {
                            delete holder;
                            delete _call;
                            delete _sa;

                            _this->proc_lock.lock();
                            try
                            {
                                _action(data);
                                if ( my_pid->is_valid() ) _this->proc_lock.unlock();
                            }
                            catch(...)
                            {
                                log::post(LOG_ERROR, "Exception caught in unprotect() continuation");
                                _this->fault(Exception);
                            }
                        }), nullptr);

                    data_lock.unlock();
                });

            _holder[0] = (void *)call;
            _holder[1] = (void *)sa;

            sa->link_target(call);
            src->link_target(sa);

            return true;
        }
    #pragma endregion

        //
        // actor Lifetime
        //

        /// <summary>
        /// The ConcRT Agents run() method, used to start the actor.
        /// </summary>
        _DISPIMP void run();

        /// <summary>
        /// All actors should override the execute method if they have messaging that should
        /// be handled using accept() calls instead of message handlers (callbacks).
        /// </summary>
        _DISPIMP virtual void execute() { }

        /// <summary>
        /// Reason code for actor initialization
        /// </summary>
        enum init_reason { ACTR_Create, ACTR_Recovered, ACTR_Respawn, ACTR_Migrate };

        /// <summary>
        /// Initialize is called after the constructor and before the actor's execute() function is called.
        /// It is usually used to set up message handlers and to perform general setup tasks. It is called
        /// both when the lifetime of an actor starts and when it is resumed after a suspension, fault, or
        /// host crash.
        /// </summary>
        _DISPIMP virtual void initialize(init_reason reason = ACTR_Create) { }

        /// <summary>
        /// Inform the host that the actor's lifetime has come to an end. All resources associated
        /// with the actor will be cleaned up, including checkpoints locally and remotely stored.
        /// </summary>
        _DISPIMP void Done(); 



        //
        // actor Checkpoint, Restore, Migration and Serialization
        //


        /// <summary>
        /// Save the internal state of the actor into a byte vector.
        /// </summary>
        _DISPIMP bool save(storage_block &block);

        /// <summary>
        /// Create a checkpoint for the process. Checkpoints are saved locally (in the process) itself
        /// as well as in any actor storage defined for the host.
        /// </summary>
        _DISPIMP bool checkpoint();

        /// <summary>
        /// Create a checkpoint, then remove the actor from the local host. The actor may be
        /// restored at a later point.
        /// </summary>
        _DISPIMP bool suspend();

        /// <summary>
        /// Create a checkpoint, then remove the actor from the local host. The actor may be
        /// restored at a later point.
        /// </summary>
        _DISPIMP bool migrate(const std::string &to);

        /// <summary>
        /// Restore the actor from its latest locally stored checkpoint.
        /// </summary>
        _DISPIMP bool restore();

        /// <summary>
        /// Restore the actor from a specific buffer. Used when a client is responsible for
        /// keeping the state of the actor between invocations.
        /// </summary>
        _DISPIMP bool restore(storage_block &block);

        /// <summary>
        /// actor serialization base code.
        /// </summary>
        _DISPIMP virtual size_t calculate_size() const;
        _DISPIMP virtual bool serialize(actors::Serializer &serializer) const;
        _DISPIMP virtual bool deserialize(actors::Serializer &serializer);

        

        //
        // Miscellaneous Protected Functions
        //

        /// <summary>
        /// Allow a block of code to not hold the internal actor lock. This is particularly useful
        /// when doing blocking I/O from within the execute() method or some message handler.
        /// </summary>
        template<typename Func1, typename Func2>
        void unprotect(Func1 func, Func2 continuation)
        {
            pplx::create_task([=,this]()
            {
                Concurrency::Context::Oversubscribe(true);
               
                func();

                Concurrency::Context::Oversubscribe(false);

                auto _this = this;
                auto my_pid = self_pid;

                data_lock.lock();

                _schedule(CreateContinuation(
                    [=]()
                    {
                        _this->proc_lock.lock();
                        try
                        {
                            continuation();
                            if ( my_pid->is_valid() ) _this->proc_lock.unlock();
                        }
                        catch(...)
                        {
                            log::post(LOG_ERROR, "Exception caught in unprotect() continuation");
                            fault(Exception);
                        }
                    }), nullptr);

                data_lock.unlock();
            });

            _run_scheduled();
        }

        /// <summary>
        /// Each actor may retrieve its own PID through this function. Having
        /// the "self identifier" is useful for sending as a 'reply_to' argument in
        /// messages.
        /// </summary>
        _DISPIMP PID self() const { return *self_pid; }

        /// <summary>
        /// Send a fault message.
        /// </summary>
        _DISPIMP void fault(exit_reason reason);

        _DISPIMP IMessageDispatcher channel() const { return self_pid->m_actor_info ? self_pid->m_actor_info->chan : nullptr; }

    #pragma region "link()"


        //
        //                          LINK
        //


        /// <summary>
        /// Linking means to register to listen to faults coming from the process
        /// identified by a PID. Linking may be symmetrical, in which the linking
        /// goes both ways, or asymmetrical, in which case the other actor is
        /// not getting notified of faults of the initiating process.
        /// </summary>
        _DISPIMP bool link(const PID other, bool symmetrical = true);

        // TFS: # 336228
        // TODO: Support UNLINK
        // TFS: # 336229
        // TODO: Keep a registry of linked processed in the communication layer so that 
        //       faults can be propagated when a TCP connection goes down.

        /// <summary>
        /// Spawn-and-link will atomically spawn a new actor and link to it. This
        /// avoid the race inherent in first spawning an actor (which is started) and
        /// then linking to it. The link may be either symmetrical or asymmetrical.
        /// </summary>
        _DISPIMP PID spawn_and_link(const std::string &uri, bool symmetrical = true);
        _DISPIMP PID spawn_and_link(const std::string &baseUri, const std::string &resource, bool symmetrical = true);

        /// <summary>
        /// Spawn-and-link will atomically spawn a new actor and link to it. This
        /// avoid the race inherent in first spawning an actor (which is started) and
        /// then linking to it. The link may be either symmetrical or asymmetrical.
        /// </summary>
        template<typename ActorType>
        PID spawn_and_link(const std::string &uri_string = "", bool symmetrical = true) 
        {
            http::uri uri(uri_string);

            // the uri should identify a host only
            if (!uri.empty() && uri.is_authority())
            {
                return PID().invalidate();
            }

            tcp_listener *listener = tcp_listener::find(uri);

            if ( listener != nullptr )
            {
                actor *proc = reinterpret_cast<actor *>(new ActorType);
                actor_info p_info = proc->self_pid->m_actor_info;
                if ( !p_info ) return PID();

                tcp_listener::_register_actor(p_info->guid, false, p_info);

                if (symmetrical) _add_link(proc->self());
                proc->_add_link(self());

                process_factory factory = listener->_get_factory(typeid(ActorType).name());
                if ( factory ) proc->set_context(listener, factory->Context);

                proc->initialize(ACTR_Create);
                proc->start();
            
                return PID(p_info);
            }
        
            return PID(_actor_info::_spawn_link_remote(typeid(ActorType).name(), self(), uri));
        }
    #pragma endregion

    #pragma region "accept()"


        //
        //                          ACCEPT
        //


        /// <summary>
        /// Accept is the counterpart of 'send,' within the process. It is
        /// how messages are received and returned to the controlling code.
        ///
        /// Each actor has its own single inbox, into which messages are
        /// delivered and maintained in the order they arrive. It is not
        /// possible to accept a message that isn't at the head of the
        /// inbox queue.
        /// </summary>
        /// <summary>
        /// Type-safe message accept
        /// </summary>
        /// <remarks>
        /// Messages are separated by their type and the system uses strict
        /// type equivalence to determine a match. No conversion or inheritance
        /// relationships are considered in matching accepts with messages
        /// in the inbox: a waiting int cannot, for example, be received
        /// with an accept<long>(l), even though ints are implicitly converted
        /// to longs.
        /// </remarks>
        _DISPIMP pplx::task<general_msg *> accept();

        template<typename Payload>
        pplx::task<Payload> accept()
        {
            _validate();

            const char *type_name = typeid(Payload).name();
            const char *tname = typeid(Payload).name();

            data_lock.lock();

            if ( msg_inbox.size() > 0 )
            {
                general_msg *msg = msg_inbox.front();

                if ( msg != nullptr && msg->Matches(type_name) )
                {
                    msg_inbox.pop();

                    Payload data;

                    bool result = _actor_info::process_message(msg, data);
                    delete msg;

                    data_lock.unlock();

                    if ( result ) return pplx::create_value(data, *this);
                }
            }
                   
            auto my_pid = self_pid;
            auto result = pplx::task_completion_event<Payload>(*this);
                   
            auto forwarder = CreateForwarder(true, [this,my_pid,result](general_msg *msg) -> bool
            {
                auto r = result;

                Payload data;

                // Silently consume messages if not valid
                if ( msg == nullptr || !my_pid->is_valid() || !_actor_info::process_message(msg, data) ) 
                {
                    delete msg;
                    return false;
                }

                r.set(data);

                delete msg;
                return true;
            });

            _add_waiting_accept(type_name, forwarder);

            data_lock.unlock();
            return pplx::task<Payload>(result,*this);
        }

        /// <summary>
        /// Type-safe message accept of two elements from a pair
        /// </summary>
        template<typename Payload0, typename Payload1>
        pplx::task<std::tuple<Payload0,Payload1>> accept()
        {
            _validate();

            const type_info &typeId = typeid(std::tuple<Payload0,Payload1>);

            const char *type_name = typeId.name();
            const char *tname = typeId.name();


            data_lock.lock();

            if ( msg_inbox.size() > 0 )
            {
                general_msg *msg = msg_inbox.front();

                if ( msg != nullptr && msg->Matches(type_name) )
                {
                    msg_inbox.pop();

                    std::tuple<Payload0,Payload1> data;

                    bool result = _actor_info::process_message(msg, data);
                    delete msg;

                    data_lock.unlock();

                    if ( result ) return pplx::create_value(data, *this);
                }
            }
                   
            auto my_pid = self_pid;
            auto result = pplx::task_completion_event<std::tuple<Payload0,Payload1>>(*this);
                   
            auto forwarder = CreateForwarder(true, [this,my_pid,result](general_msg *msg) -> bool
            {
                auto r = result;

                std::tuple<Payload0,Payload1> data;

                // Silently consume messages if not valid
                if ( msg == nullptr || !my_pid->is_valid() || !_actor_info::process_message(msg, data) ) 
                {
                    delete msg;
                    return false;
                }

                r.set(data);

                delete msg;
                return true;
            });

            _add_waiting_accept(type_name, forwarder);

            data_lock.unlock();

            return pplx::task<std::tuple<Payload0,Payload1>>(result,*this);
        }

        /// <summary>
        /// Type-safe message accept of three elements from a triple
        /// </summary>
        template<typename Payload0, typename Payload1, typename Payload2>
        pplx::task<std::tuple<Payload0,Payload1,Payload2>> accept()
        {
            _validate();

            const type_info &typeId = typeid(std::tuple<Payload0,Payload1,Payload2>);

            const char *type_name = typeId.name();
            const char *tname = typeId.name();


            data_lock.lock();

            if ( msg_inbox.size() > 0 )
            {
                general_msg *msg = msg_inbox.front();

                if ( msg != nullptr && msg->Matches(type_name) )
                {
                    msg_inbox.pop();

                    std::tuple<Payload0,Payload1,Payload2> data;

                    bool result = _actor_info::process_message(msg, data);
                    delete msg;

                    data_lock.unlock();

                    if ( result ) return pplx::create_value(data, *this);
                }
            }
                   
            auto my_pid = self_pid;
            auto result = pplx::task_completion_event<std::tuple<Payload0,Payload1,Payload2>>(*this);
                   
            auto forwarder = CreateForwarder(true, [this,my_pid,result](general_msg *msg) -> bool
            {
                auto r = result;

                std::tuple<Payload0,Payload1,Payload2> data;

                // Silently consume messages if not valid
                if ( msg == nullptr || !my_pid->is_valid() || !_actor_info::process_message(msg, data) ) 
                {
                    delete msg;
                    return false;
                }

                r.set(data);

                delete msg;
                return true;
            });

            _add_waiting_accept(type_name, forwarder);

            data_lock.unlock();

            return pplx::task<std::tuple<Payload0,Payload1,Payload2>>(result,*this);
        }

        /// <summary>
        /// Type-safe message accept of four elements from a quadruple
        /// </summary>
        template<typename Payload0, typename Payload1, typename Payload2, typename Payload3>
        pplx::task<std::tuple<Payload0,Payload1,Payload2,Payload3>> accept()
        {
            _validate();

            const type_info &typeId = typeid(std::tuple<Payload0,Payload1,Payload2,Payload3>);

            const char *type_name = typeId.name();
            const char *tname = typeId.name();


            data_lock.lock();

            if ( msg_inbox.size() > 0 )
            {
                general_msg *msg = msg_inbox.front();

                if ( msg != nullptr && msg->Matches(type_name) )
                {
                    msg_inbox.pop();

                    std::tuple<Payload0,Payload1,Payload2,Payload3> data;

                    bool result = _actor_info::process_message(msg, data);
                    delete msg;

                    data_lock.unlock();

                    if ( result ) return pplx::create_value(data, *this);
                }
            }
                   
            auto my_pid = self_pid;
            auto result = pplx::task_completion_event<std::tuple<Payload0,Payload1,Payload2,Payload3>>(*this);
                   
            auto forwarder = CreateForwarder(true, [this,my_pid,result](general_msg *msg) -> bool
            {
                auto r = result;

                std::tuple<Payload0,Payload1,Payload2,Payload3> data;

                // Silently consume messages if not valid
                if ( msg == nullptr || !my_pid->is_valid() || !_actor_info::process_message(msg, data) ) 
                {
                    delete msg;
                    return false;
                }

                r.set(data);

                delete msg;
                return true;
            });

            _add_waiting_accept(type_name, forwarder);

            data_lock.unlock();
            return pplx::task<std::tuple<Payload0,Payload1,Payload2,Payload3>>(result,*this);
        }
    #pragma endregion

    #pragma region "forward()"


        //
        //                          FORWARD
        //


        //
        // While accept is used in code to receive data in a blocking fashion from
        // within control-flow constructs, the fact that it is a blocking API places
        // some limits on scalability of code using it. To address this, actors that
        // have portions that are largely stateless may instead handle messages via
        // callbacks, or by passing messages into ITarget objects.
        //
    
        /// <summary>
        /// Forward all messages of type 'Payload' to the specified target.
        /// </summary>
        template<typename Payload>
        bool forward(Concurrency::ITarget<Payload> *target)
        {
            if ( target == nullptr ) return false;

            _validate();

            std::string type_name = typeid(Payload).name();

            data_lock.lock();
            if (attached_dispatchers[type_name] != NULL) { data_lock.unlock(); return false; }

            auto my_pid = self_pid;

            attached_dispatchers[type_name] = CreateForwarder(false, [this,my_pid,target](general_msg *msg) -> bool
            {
                // Silently consume messages if not valid
                if ( msg == nullptr || !my_pid->is_valid() ) return false;

                Payload data;
                if ( _actor_info::process_message(msg, data) )
                    Concurrency::asend(target, data);
                delete msg;
                return true;
            });
            data_lock.unlock();

            return true;
        }

        /// <summary>
        /// Forward all messages of type 'Payload' to the specified target.
        /// </summary>
        template<typename Payload>
        bool forward(Concurrency::ITarget<Payload> &target)
        {
            return forward<Payload>(&target);
        }

        /// <summary>
        /// Forward all messages of type 'Payload' to the specified functor.
        /// </summary>
        template<typename Payload,typename Functor>
        bool forward(const Functor &func)
        {
            _validate();

            std::string type_name = typeid(Payload).name();

            data_lock.lock();
            if (attached_dispatchers[type_name] != NULL) { data_lock.unlock(); return false; };

            auto my_pid = self_pid;

            attached_dispatchers[type_name] = CreateForwarder(false, [this,my_pid,func](general_msg *msg) -> bool
            {
                // Silently consume messages if not valid
                if ( msg == nullptr || !my_pid->is_valid() ) 
                {
                    if ( msg != nullptr && msg->flags & MSGFLG_TRACE ) printf("    Forward: INVALID\n");
                    delete msg;
                    return false;
                }


                Payload data;

                if ( _actor_info::process_message(msg, data) )
                {
                    proc_lock.lock();

                    if ( my_pid->is_valid() )
                    {
                        set_current(this);
                        try
                        {
                            func(data);
                            if ( my_pid->is_valid() ) proc_lock.unlock();
                        }
                        catch(...)
                        {
                            //fprintf(stderr, "Exception caught @%d\n", 2);fflush(stderr);
                            fault(Exception);
                        }
                        set_current(nullptr);
                    }
                }
                delete msg;
                return true;
            });

            data_lock.unlock();
            return true;
        }

        /// <summary>
        /// Forward all messages of that are pairs of types 'Payload0' and 'Payload1'
        /// to the specified functor.
        /// </summary>
        template<typename Payload0,typename Payload1,typename Functor>
        bool forward(const Functor &func)
        {
            _validate();

            std::string type_name = typeid(std::tuple<Payload0,Payload1>).name();

            data_lock.lock();
            if (attached_dispatchers[type_name] != NULL) { data_lock.unlock(); return false; };

            auto my_pid = self_pid;

            attached_dispatchers[type_name] = CreateForwarder(false, [this,my_pid,func](general_msg *msg) -> bool
            {
                // Silently consume messages if not valid
                if ( msg == nullptr || !my_pid->is_valid() )
                {
                    if ( msg != nullptr && msg->flags & MSGFLG_TRACE ) printf("    Forward: INVALID\n");
                    delete msg;
                    return false;
                }

                std::tuple<Payload0,Payload1> data;
                if (_actor_info::process_message(msg, data))
                {
                    proc_lock.lock();

                    if ( my_pid->is_valid() )
                    {
                        set_current(this);
                        try
                        {
                            func(std::get<0>(data), std::get<1>(data));
                            if ( my_pid->is_valid() ) proc_lock.unlock();
                        }
                        catch(...)
                        {
                            //fprintf(stderr, "Exception caught @%d\n", 3);
                            this->fault(Exception);
                        }
                        set_current(nullptr);
                    }
                }
                delete msg;
                return true;
            });

            data_lock.unlock();
            return true;
        }

        /// <summary>
        /// Forward all messages of that are triples of types 'Payload0,' 'Payload1,' and 'Payload2'
        /// to the specified functor.
        /// </summary>
        template<typename Payload0,typename Payload1,typename Payload2,typename Functor>
        bool forward(const Functor &func)
        {
            _validate();

            std::string type_name = typeid(std::tuple<Payload0,Payload1,Payload2>).name();

            data_lock.lock();
            if (attached_dispatchers[type_name] != NULL) { data_lock.unlock(); return false; };

            auto my_pid = self_pid;

            attached_dispatchers[type_name] = CreateForwarder(false, [my_pid,this,func](general_msg *msg) -> bool
            {
                // Silently consume messages if not valid
                if ( msg == nullptr || !my_pid->is_valid() )
                {
                    if ( msg != nullptr && msg->flags & MSGFLG_TRACE ) printf("    Forward: INVALID\n");
                    delete msg;
                    return false;
                }

                std::tuple<Payload0,Payload1,Payload2> data;
                if ( _actor_info::process_message(msg, data) )
                {
                    proc_lock.lock();

                    if ( my_pid->is_valid() )
                    {
                        set_current(this);
                        try
                        {
                            func(std::get<0>(data), std::get<1>(data), std::get<2>(data));
                            if ( my_pid->is_valid() ) proc_lock.unlock();
                        }
                        catch(...)
                        {
                            //fprintf(stderr, "Exception caught @%d\n", 3);
                            fault(Exception);
                        }
                        set_current(nullptr);
                    }
                }
                delete msg;
                return true;

            });

            data_lock.unlock();
            return true;
        }

        /// <summary>
        /// Forward all messages of that are quadruples of types 'Payload0,' 'Payload1,' 'Payload2,' and 'Payload3'
        /// to the specified functor.
        /// </summary>
        template<typename Payload0,typename Payload1,typename Payload2,typename Payload3,typename Functor>
        bool forward(const Functor &func)
        {
            _validate();

            std::string type_name = typeid(std::tuple<Payload0,Payload1,Payload2,Payload3>).name();

            data_lock.lock();
            if (attached_dispatchers[type_name] != NULL) { data_lock.unlock(); return false; };

            auto my_pid = self_pid;

            attached_dispatchers[type_name] = CreateForwarder(false, [my_pid,this,func](general_msg *msg) -> bool
            {
                // Silently consume messages if not valid
                if ( msg == nullptr || !my_pid->is_valid() )
                {
                    if ( msg != nullptr && msg->flags & MSGFLG_TRACE ) printf("    Forward: INVALID\n");
                    delete msg;
                    return false;
                }

                std::tuple<Payload0,Payload1,Payload2,Payload3> data;
                if ( _actor_info::process_message(msg, data) )
                {
                    proc_lock.lock();

                    if ( my_pid->is_valid() )
                    {
                        set_current(this);
                        try
                        {
                            func(std::get<0>(data), std::get<1>(data), std::get<2>(data), std::get<3>(data));
                            if ( my_pid->is_valid() ) proc_lock.unlock();
                        }
                        catch(...)
                        {
                            //fprintf(stderr, "Exception caught @%d\n", 3);
                            fault(Exception);
                        }
                        set_current(nullptr);
                    }
                }
                delete msg;
                return true;

            });

            data_lock.unlock();
            return true;
        }

        /// <summary>
        /// Forward all request messages of type 'ReqT/RepT' to the specified functor. The
        /// reply is automatically sent back based on the return value of the functor
        /// invocation.
        /// </summary>
        template<typename ReqT, typename RepT, typename Functor>
        bool forward_request(const Functor &func)
        {
            _validate();

            std::string type_name = typeid(reqrep_msg<ReqT,RepT>).name();

            data_lock.lock();
            if (attached_dispatchers[type_name] != NULL) { data_lock.unlock(); return false; }

            auto my_pid = self_pid;

            auto call = CreateForwarder(false, [my_pid,this,func](general_msg *msg) -> bool
            {
                // Silently consume messages if not valid
                if ( msg == nullptr || !my_pid->is_valid() ) 
                {
                    if ( msg != nullptr && msg->flags & MSGFLG_TRACE ) printf("    forward_request: INVALID\n");
                    delete msg;
                    return false;
                }

                reqrep_msg<ReqT,RepT> data;
                if ( _actor_info::process_message(msg, data) )
                {
                    proc_lock.lock();

                    if ( my_pid->is_valid() )
                    {
                        set_current(this);
                        try
                        {
                            func(data.reply_to, data.data);
                            if ( my_pid->is_valid() ) proc_lock.unlock();
                        }
                        catch(...)
                        {
                            //fprintf(stderr, "Exception caught @%d\n", 5);
                            fault(Exception);
                        }
                        set_current(nullptr);
                    }
                }
                delete msg;
                return true;
            });

            attached_dispatchers[type_name] = call;
            data_lock.unlock();
            return true;
        }

        /// <summary>
        /// Forward all faults to the specified functor, which will be
        /// invoked from within a call block holding the actor lock. This is
        /// the only way to get faults, they cannot be received with a call to
        /// accept().
        /// </summary>
        /// <remarks>
        /// Faults may be forwarded to a function that accepts two or three arguments:
        /// the third is an optional reference to the checkpointed state (if any) of
        /// the faulted process.
        /// </remarks>
        template<typename Functor>
        bool forward_fault(const Functor &func)
        {
            _validate();

            std::string type_name = typeid(fault_msg).name();

            data_lock.lock();

            auto my_pid = self_pid;

            attached_dispatchers[type_name] = CreateForwarder(false, [my_pid,this,func](general_msg *msg) -> bool
            {
                // Silently consume messages if not valid
                if ( msg == nullptr || !my_pid->is_valid() )
                {
                    if ( msg != nullptr && msg->flags & MSGFLG_TRACE ) printf("    forward_fault: INVALID\n");
                    delete msg;
                    return false;
                }

                fault_msg data;
                if (_actor_info::process_message(msg, data))
                {
                    PID faulted_pid = data->Faulted;
                    msg_guid faulted = faulted_pid.m_actor_info->guid;

                    data_lock.lock();
                    auto pos = linked_processes.begin();
                    for (; pos != linked_processes.end(); pos++)
                    {
                        if (pos->m_actor_info->guid == faulted)
                            break;
                    }

                    if (pos != linked_processes.end())
                        linked_processes.erase(pos);
                
                    data_lock.unlock();

                    proc_lock.lock();

                    if ( (data->Reason & (Migration | Suspension)) == 0 )
                        faulted_pid.m_actor_info->faulted = true;

                    if ( my_pid->is_valid() )
                    {
                        set_current(this);
                        try
                        {
                            func(data);
                            if ( my_pid->is_valid() ) proc_lock.unlock();
                        }
                        catch(...)
                        {
                            log::post(LOG_ERROR, "Exception caught");
                            fault(Exception);
                        }
                        set_current(nullptr);
                    }
                }
                delete msg;
                return true;
            });

            data_lock.unlock();

            return true;
        }
    #pragma endregion

    #pragma region "request()"


        //
        //                          REQUEST
        //


        /// <summary>
        /// A version of Request that should be used within actor bodies in place of PID::Request()
        /// </summary>
        template<typename ReqT, typename RepT, typename Functor>
        bool request(PID pid, const ReqT &input, Functor func)
        {
            if ( !pid.is_valid() || !pid.m_actor_info->chan ) return false;

            bool result = false;
        
            Concurrency::overwrite_buffer<RepT> *reply = new Concurrency::overwrite_buffer<RepT>();

            PID replyPID = PID::forward(reply, true);

            reqrep_msg<ReqT,RepT> msg;
            msg.data = input;
            msg.reply_to = replyPID;

            result = pid.send(msg) && 
                     receive<RepT>((Concurrency::ISource<RepT> *)reply, 
                                   [=](RepT data) { PID rPID = replyPID; func(data); rPID.invalidate(); delete reply; });

            if ( !result )
            {
                replyPID.invalidate();
                delete reply;
            }

            return result;
        }

        template<typename ReqT_1, typename ReqT_2, typename RepT, typename Functor>
        bool request(PID pid, const ReqT_1 &input1, const ReqT_2 &input2, Functor func)
        {
            typedef reqrep_msg<std::tuple<ReqT_1,ReqT_2>,RepT> msg_type;

            if ( !pid.is_valid() || !pid.m_actor_info->chan ) return false;

            bool result = false;
        
            Concurrency::overwrite_buffer<RepT> *reply = new Concurrency::overwrite_buffer<RepT>();

            PID replyPID = PID::forward(reply, true);

            msg_type msg;
            msg.data = std::make_tuple(input1,input2);
            msg.reply_to = replyPID;

            result = pid.send<msg_type>(msg);

            result = pid.send(msg) && 
                     receive<RepT>((Concurrency::ISource<RepT> *)reply, 
                                   [=](RepT data) { PID rPID = replyPID; func(data); rPID.invalidate(); delete reply; });

            if ( !result )
            {
                replyPID.invalidate();
                delete reply;
            }

            return result;
        }

        template<typename ReqT_1, typename ReqT_2, typename ReqT_3, typename RepT, typename Functor>
        bool request(PID pid, const ReqT_1 &input1, const ReqT_2 &input2, const ReqT_3 &input3, Functor func)
        {
            typedef reqrep_msg<std::tuple<ReqT_1,ReqT_2,ReqT_3>,RepT> msg_type;

            if ( !pid.is_valid() || !pid.m_actor_info->chan ) return false;

            bool result = false;
        
            Concurrency::overwrite_buffer<RepT> *reply = new Concurrency::overwrite_buffer<RepT>();

            PID replyPID = PID::forward(reply, true);

            msg_type msg;
            msg.data = std::make_tuple(input1,input2,input3);
            msg.reply_to = replyPID;

            result = pid.send<msg_type>(msg);

            result = pid.send(msg) && 
                     receive<RepT>((Concurrency::ISource<RepT> *)reply, 
                                   [=](RepT data) { PID rPID = replyPID; func(data); rPID.invalidate(); delete reply; });

            if ( !result )
            {
                replyPID.invalidate();
                delete reply;
            }

            return result;
        }

        template<typename ReqT_1, typename ReqT_2, typename ReqT_3, typename ReqT_4, typename RepT, typename Functor>
        bool request(PID pid, const ReqT_1 &input1, const ReqT_2 &input2, const ReqT_3 &input3, const ReqT_4 &input4, Functor func)
        {
            typedef reqrep_msg<std::tuple<ReqT_1,ReqT_2,ReqT_3,ReqT_4>,RepT> msg_type;

            if ( !pid.is_valid() || !pid.m_actor_info->chan ) return false;

            bool result = false;
        
            Concurrency::overwrite_buffer<RepT> *reply = new Concurrency::overwrite_buffer<RepT>();

            PID replyPID = PID::forward(reply, true);

            msg_type msg;
            msg.data = std::make_tuple(input1,input2,input3,input4);
            msg.reply_to = replyPID;

            result = pid.send<msg_type>(msg);

            result = pid.send(msg) && 
                     receive<RepT>((Concurrency::ISource<RepT> *)reply, 
                                   [=](RepT data) { PID rPID = replyPID; func(data); rPID.invalidate(); delete reply; });

            if ( !result )
            {
                replyPID.invalidate();
                delete reply;
            }

            return result;
        }

        /// <summary>
        /// Requests are accepted within actor bodies using AcceptRequest. The PID passed out is where a reply should be
        /// sent. A actor accepting a request should expect that it must eventually send a reply, as the other party is
        /// likely blocked waiting for it.
        /// </summary>
        template<typename ReqT, typename RepT>
        pplx::task<std::tuple<ReqT,PID>> accept_request()
        {
            auto acc = accept<reqrep_msg<ReqT,RepT>>();

            if ( acc.is_done() )
            {
                reqrep_msg<ReqT,RepT> msg = acc.get();
                return pplx::create_value(std::make_tuple(msg.data, msg.reply_to),*this);            
            } 

            pplx::task_completion_event<std::tuple<ReqT,PID>> result(*this);

            acc.then(
                [=](pplx::task<reqrep_msg<ReqT,RepT>> op)
                {
                    auto r = result; 
                    auto msg = op.get();
                    r.set(std::make_tuple(msg.data, msg.reply_to)); 
                });

            return pplx::task<std::tuple<ReqT,PID>>(result,*this);
        }

        /// <summary>
        /// Requests are accepted within actor bodies using AcceptRequest. The PID passed out is where a reply should be
        /// sent. A actor accepting a request should expect that it must eventually send a reply, as the other party is
        /// likely blocked waiting for it.
        /// </summary>
        template<typename Payload0, typename Payload1, typename RepT>
        pplx::task<std::tuple<Payload0, Payload1, PID>> accept_request()
        {
            auto acc = accept<reqrep_msg<std::tuple<Payload0,Payload1>,RepT>>();

            if ( acc.is_done() )
            {
                reqrep_msg<std::tuple<Payload0,Payload1>,RepT> msg = acc.get();
                return pplx::create_value(std::make_tuple(std::get<0>(msg.data), std::get<1>(msg.data), msg.reply_to),*this);
            
            } 

            pplx::task_completion_event<std::tuple<Payload0, Payload1, PID>> result(*this);

            acc.then(
                [=](pplx::task<reqrep_msg<std::tuple<Payload0,Payload1>,RepT>> op)
                {
                    auto r = result; 
                    auto msg = op.get();
                    r.set(std::make_tuple(std::get<0>(msg.data), std::get<1>(msg.data), msg.reply_to)); 
                });

            return pplx::task<std::tuple<Payload0, Payload1, PID>>(result,*this);
        }

        /// <summary>
        /// Requests are accepted within actor bodies using AcceptRequest. The PID passed out is where a reply should be
        /// sent. A actor accepting a request should expect that it must eventually send a reply, as the other party is
        /// likely blocked waiting for it.
        /// </summary>
        template<typename Payload0, typename Payload1, typename Payload2, typename RepT>
        pplx::task<std::tuple<Payload0, Payload1, Payload2, PID>> accept_request()
        {
            auto acc = accept<reqrep_msg<std::tuple<Payload0,Payload1,Payload2>,RepT>>();

            if ( acc.is_done() )
            {
                reqrep_msg<std::tuple<Payload0,Payload1,Payload2>,RepT> msg = acc.get();
                return pplx::create_value(std::make_tuple(std::get<0>(msg.data), std::get<1>(msg.data), std::get<2>(msg.data), msg.reply_to),*this);           
            } 

            pplx::task_completion_event<std::tuple<Payload0, Payload1, Payload2, PID>> result(*this);

            acc.then(
                [=](pplx::task<reqrep_msg<std::tuple<Payload0,Payload1,Payload2>,RepT>> op)
                {
                    auto r = result; 
                    auto msg = op.get();
                    r.set(std::make_tuple(std::get<0>(msg.data), std::get<1>(msg.data), std::get<2>(msg.data), msg.reply_to)); 
                });

            return pplx::task<std::tuple<Payload0, Payload1, Payload2, PID>>(result,*this);
        }

        /// <summary>
        /// Requests are accepted within actor bodies using AcceptRequest. The PID passed out is where a reply should be
        /// sent. A actor accepting a request should expect that it must eventually send a reply, as the other party is
        /// likely blocked waiting for it.
        /// </summary>
        template<typename Payload0, typename Payload1, typename Payload2, typename Payload3, typename RepT>
        pplx::task<std::tuple<Payload0, Payload1, Payload2, Payload3, PID>> accept_request()
        {
            auto acc = accept<reqrep_msg<std::tuple<Payload0,Payload1,Payload2,Payload3>,RepT>>();

            if ( acc.is_done() )
            {
                reqrep_msg<std::tuple<Payload0,Payload1,Payload2,Payload3>,RepT> msg = acc.get();
                return pplx::create_value(std::make_tuple(std::get<0>(msg.data), std::get<1>(msg.data), std::get<2>(msg.data), std::get<3>(msg.data), msg.reply_to),*this);           
            } 

            pplx::task_completion_event<std::tuple<Payload0, Payload1, Payload2, Payload3, PID>> result(*this);

            acc.then(
                [=](pplx::task<reqrep_msg<std::tuple<Payload0,Payload1,Payload2,Payload3>,RepT>> op)
                {
                    auto r = result; 
                    auto msg = op.get();
                    r.set(std::make_tuple(std::get<0>(msg.data), std::get<1>(msg.data), std::get<2>(msg.data), std::get<3>(msg.data), msg.reply_to)); 
                });

            return pplx::task<std::tuple<Payload0, Payload1, Payload2, Payload3, PID>>(result,*this);
        }

#pragma endregion

    #pragma region "Choose()"

        //
        //                          CHOOSE
        //

        /// <summary>
        /// choose will wait for the first of two possible input data types 
        /// to arrive and will return 
        /// </summary>
        /// <typeparam name="Payload0">
        /// The first payload type 
        /// </typeparam>
        /// <typeparam name="Payload1">
        /// The second payload type 
        /// </typeparam>
        /// <param name="data0">
        /// The first reference used to capture data.
        /// </param>
        /// <param name="data1">
        /// The second reference used to capture data.
        /// </param>
        /// <returns>
        /// The (zero-based) index of the item it picked up. In other words, 
        /// if data0 is filled in, the return value will be 0, otherwise 1.
        /// </returns>
        template<typename Payload0, typename Payload1, typename Func0, typename Func1>
        int choose(Payload0 &data0, Payload1 &data1, Func0 &action0, Func1 &action1)
        {
            _validate();

            int idx = -1;

            const char *type_name_0 = typeid(Payload0).name();
            const char *type_name_1 = typeid(Payload1).name();

            data_lock.lock();

            if ( msg_inbox.size() > 0 )
            {
                general_msg *msg = msg_inbox.front();

                if ( msg != nullptr && msg->Matches(type_name_0) )
                    idx = 0;
                else if ( msg != nullptr && msg->Matches(type_name_1) )
                    idx = 1;

                switch (idx)
                {
                case 0:
                    {
                        msg_inbox.pop();

                        bool result = _actor_info::process_message(msg, data0);
                        delete msg;

                        data_lock.unlock();

                        if ( result ) action0(true, data0);

                        return result ? 0 : -1;
                    }
                case 1:
                    {
                        msg_inbox.pop();

                        bool result = _actor_info::process_message(msg, data1);
                        delete msg;

                        data_lock.unlock();

                        if ( result ) action1(true, data1);

                        return result ? 1 : -1;
                    }
                }
            }
                   
            auto my_pid = self_pid;

            auto forwarder = CreateForwarder(true, [=,this](general_msg *msg) -> bool
            {
                // Silently consume messages if not valid
                if ( msg == nullptr || !my_pid->is_valid() ) 
                {
                    delete msg;
                    return false;
                }

                int idx = -1;

                if ( msg->Matches(type_name_0) )
                    idx = 0;
                else if ( msg->Matches(type_name_1) )
                    idx = 1;

                switch (idx)
                {
                case 0:
                    Payload0 data0;

                    if ( _actor_info::process_message(msg, data0) )
                    {
                        proc_lock.lock();

                        if ( my_pid->is_valid() )
                            try
                            {
                                action0(false, data0);
                                if ( my_pid->is_valid() ) proc_lock.unlock();
                            }
                            catch(...)
                            {
                                log::post(LOG_ERROR, "Exception caught in accept callback");
                                fault(Exception);
                            }
                    }
                    break;
                case 1:
                    Payload1 data1;

                    if ( _actor_info::process_message(msg, data1) )
                    {
                        proc_lock.lock();

                        if ( my_pid->is_valid() )
                            try
                            {
                                action1(false, data1);
                                if ( my_pid->is_valid() ) proc_lock.unlock();
                            }
                            catch(...)
                            {
                                log::post(LOG_ERROR, "Exception caught in accept callback");
                                fault(Exception);
                            }
                    }
                    break;
                }

                delete msg;
                return true;
            });

            _add_waiting_accept(type_name_0, forwarder);
            _add_waiting_accept(type_name_1, forwarder);

            data_lock.unlock();

            return -1;
        }

        /// <summary>
        /// choose will wait for the first of three possible input data types 
        /// to arrive and will return 
        /// </summary>
        /// <typeparam name="Payload0">
        /// The first payload type 
        /// </typeparam>
        /// <typeparam name="Payload1">
        /// The second payload type 
        /// </typeparam>
        /// <typeparam name="Payload2">
        /// The third payload type 
        /// </typeparam>
        /// <param name="data0">
        /// The first reference used to capture data.
        /// </param>
        /// <param name="data1">
        /// The second reference used to capture data.
        /// </param>
        /// <param name="data2">
        /// The third reference used to capture data.
        /// </param>
        /// <returns>
        /// The (zero-based) index of the item it picked up. In other words, 
        /// if data0 is filled in, the return value will be 0, if data1 is filled
        /// in, it will be 1, etc..
        /// </returns>
        template<typename Payload0, typename Payload1, typename Payload2, typename Func0, typename Func1, typename Func2>
        int choose(Payload0 &data0, Payload1 &data1, Payload2 &data2, Func0 &action0, Func1 &action1, Func2 &action2)
        {
            _validate();

            int idx = -1;

            const char *type_name_0 = typeid(Payload0).name();
            const char *type_name_1 = typeid(Payload1).name();
            const char *type_name_2 = typeid(Payload2).name();

            data_lock.lock();

            if ( msg_inbox.size() > 0 )
            {
                general_msg *msg = msg_inbox.front();

                if ( msg != nullptr && msg->Matches(type_name_0) )
                    idx = 0;
                else if ( msg != nullptr && msg->Matches(type_name_1) )
                    idx = 1;
                else if ( msg != nullptr && msg->Matches(type_name_2) )
                    idx = 2;

                switch (idx)
                {
                case 0:
                    {
                        msg_inbox.pop();

                        bool result = _actor_info::process_message(msg, data0);
                        delete msg;

                        data_lock.unlock();

                        if ( result ) action0(true, data0);

                        return result ? 0 : -1;
                    }
                case 1:
                    {
                        msg_inbox.pop();

                        bool result = _actor_info::process_message(msg, data1);
                        delete msg;

                        data_lock.unlock();

                        if ( result ) action1(true, data1);

                        return result ? 1 : -1;
                    }
                case 2:
                    {
                        msg_inbox.pop();

                        bool result = _actor_info::process_message(msg, data2);
                        delete msg;

                        data_lock.unlock();

                        if ( result ) action2(true, data2);

                        return result ? 2 : -1;
                    }
                }
            }
                   
            auto my_pid = self_pid;

            auto forwarder = CreateForwarder(true, [=,this](general_msg *msg) -> bool
            {
                // Silently consume messages if not valid
                if ( msg == nullptr || !my_pid->is_valid() ) 
                {
                    delete msg;
                    return false;
                }

                int idx = -1;

                if ( msg->Matches(type_name_0) )
                    idx = 0;
                else if ( msg->Matches(type_name_1) )
                    idx = 1;
                else if ( msg->Matches(type_name_2) )
                    idx = 2;

                switch (idx)
                {
                case 0:
                    Payload0 data0;

                    if ( _actor_info::process_message(msg, data0) )
                    {
                        proc_lock.lock();

                        if ( my_pid->is_valid() )
                            try
                            {
                                action0(false, data0);
                                if ( my_pid->is_valid() ) proc_lock.unlock();
                            }
                            catch(...)
                            {
                                log::post(LOG_ERROR, "Exception caught in accept callback");
                                fault(Exception);
                            }
                    }
                    break;
                case 1:
                    Payload1 data1;

                    if ( _actor_info::process_message(msg, data1) )
                    {
                        proc_lock.lock();

                        if ( my_pid->is_valid() )
                            try
                            {
                                action1(false, data1);
                                if ( my_pid->is_valid() ) proc_lock.unlock();
                            }
                            catch(...)
                            {
                                log::post(LOG_ERROR, "Exception caught in accept callback");
                                fault(Exception);
                            }
                    }
                    break;
                case 2:
                    Payload2 data2;

                    if ( _actor_info::process_message(msg, data2) )
                    {
                        proc_lock.lock();

                        if ( my_pid->is_valid() )
                            try
                            {
                                action2(false, data2);
                                if ( my_pid->is_valid() ) proc_lock.unlock();
                            }
                            catch(...)
                            {
                                log::post(LOG_ERROR, "Exception caught in accept callback");
                                fault(Exception);
                            }
                    }
                    break;
                }

                delete msg;
                return true;
            });

            _add_waiting_accept(type_name_0, forwarder);
            _add_waiting_accept(type_name_1, forwarder);
            _add_waiting_accept(type_name_2, forwarder);

            data_lock.unlock();

            return -1;
        }

        /// <summary>
        /// choose will wait for the first of four possible input data types 
        /// to arrive and will return 
        /// </summary>
        /// <typeparam name="Payload0">
        /// The first payload type 
        /// </typeparam>
        /// <typeparam name="Payload1">
        /// The second payload type 
        /// </typeparam>
        /// <typeparam name="Payload2">
        /// The third payload type 
        /// </typeparam>
        /// <typeparam name="Payload3">
        /// The fourth payload type 
        /// </typeparam>
        /// <param name="data0">
        /// The first reference used to capture data.
        /// </param>
        /// <param name="data1">
        /// The second reference used to capture data.
        /// </param>
        /// <param name="data2">
        /// The third reference used to capture data.
        /// </param>
        /// <param name="data3">
        /// The fourth reference used to capture data.
        /// </param>
        /// <returns>
        /// The (zero-based) index of the item it picked up. In other words, 
        /// if data0 is filled in, the return value will be 0, if data1 is filled
        /// in, it will be 1, etc..
        /// </returns>
        template<typename Payload0, typename Payload1, typename Payload2, typename Payload3, typename Func0, typename Func1, typename Func2, typename Func3>
        int choose(Payload0 &data0, Payload1 &data1, Payload2 &data2, Payload3 &data3, Func0 &action0, Func1 &action1, Func2 &action2, Func3 &action3)
        {
            _validate();

            int idx = -1;

            const char *type_name_0 = typeid(Payload0).name();
            const char *type_name_1 = typeid(Payload1).name();
            const char *type_name_2 = typeid(Payload2).name();
            const char *type_name_3 = typeid(Payload3).name();

            data_lock.lock();

            if ( msg_inbox.size() > 0 )
            {
                general_msg *msg = msg_inbox.front();

                if ( msg != nullptr && msg->Matches(type_name_0) )
                    idx = 0;
                else if ( msg != nullptr && msg->Matches(type_name_1) )
                    idx = 1;
                else if ( msg != nullptr && msg->Matches(type_name_2) )
                    idx = 2;
                else if ( msg != nullptr && msg->Matches(type_name_3) )
                    idx = 3;

                switch (idx)
                {
                case 0:
                    {
                        msg_inbox.pop();

                        bool result = _actor_info::process_message(msg, data0);
                        delete msg;

                        data_lock.unlock();

                        if ( result ) action0(true, data0);

                        return result ? 0 : -1;
                    }
                case 1:
                    {
                        msg_inbox.pop();

                        bool result = _actor_info::process_message(msg, data1);
                        delete msg;

                        data_lock.unlock();

                        if ( result ) action1(true, data1);

                        return result ? 1 : -1;
                    }
                case 2:
                    {
                        msg_inbox.pop();

                        bool result = _actor_info::process_message(msg, data2);
                        delete msg;

                        data_lock.unlock();

                        if ( result ) action2(true, data2);

                        return result ? 2 : -1;
                    }
                case 3:
                    {
                        msg_inbox.pop();

                        bool result = _actor_info::process_message(msg, data3);
                        delete msg;

                        data_lock.unlock();

                        if ( result ) action3(true, data3);

                        return result ? 3 : -1;
                    }
                }
            }
                   
            auto my_pid = self_pid;

            auto forwarder = CreateForwarder(true, [=,this](general_msg *msg) -> bool
            {
                // Silently consume messages if not valid
                if ( msg == nullptr || !my_pid->is_valid() ) 
                {
                    delete msg;
                    return false;
                }

                int idx = -1;

                if ( msg->Matches(type_name_0) )
                    idx = 0;
                else if ( msg->Matches(type_name_1) )
                    idx = 1;
                else if ( msg->Matches(type_name_2) )
                    idx = 2;
                else if ( msg->Matches(type_name_3) )
                    idx = 3;

                switch (idx)
                {
                case 0:
                    Payload0 data0;

                    if ( _actor_info::process_message(msg, data0) )
                    {
                        proc_lock.lock();

                        if ( my_pid->is_valid() )
                            try
                            {
                                action0(false, data0);
                                if ( my_pid->is_valid() ) proc_lock.unlock();
                            }
                            catch(...)
                            {
                                log::post(LOG_ERROR, "Exception caught in accept callback");
                                fault(Exception);
                            }
                    }
                    break;
                case 1:
                    Payload1 data1;

                    if ( _actor_info::process_message(msg, data1) )
                    {
                        proc_lock.lock();

                        if ( my_pid->is_valid() )
                            try
                            {
                                action1(false, data1);
                                if ( my_pid->is_valid() ) proc_lock.unlock();
                            }
                            catch(...)
                            {
                                log::post(LOG_ERROR, "Exception caught in accept callback");
                                fault(Exception);
                            }
                    }
                    break;
                case 2:
                    Payload2 data2;

                    if ( _actor_info::process_message(msg, data2) )
                    {
                        proc_lock.lock();

                        if ( my_pid->is_valid() )
                            try
                            {
                                action2(false, data2);
                                if ( my_pid->is_valid() ) proc_lock.unlock();
                            }
                            catch(...)
                            {
                                log::post(LOG_ERROR, "Exception caught in accept callback");
                                fault(Exception);
                            }
                    }
                    break;
                case 3:
                    Payload3 data3;

                    if ( _actor_info::process_message(msg, data3) )
                    {
                        proc_lock.lock();

                        if ( my_pid->is_valid() )
                            try
                            {
                                action3(false, data3);
                                if ( my_pid->is_valid() ) proc_lock.unlock();
                            }
                            catch(...)
                            {
                                log::post(LOG_ERROR, "Exception caught in accept callback");
                                fault(Exception);
                            }
                    }
                    break;
                }

                delete msg;
                return true;
            });

            _add_waiting_accept(type_name_0, forwarder);
            _add_waiting_accept(type_name_1, forwarder);
            _add_waiting_accept(type_name_2, forwarder);
            _add_waiting_accept(type_name_3, forwarder);

            data_lock.unlock();

            return -1;
        }

        /// <summary>
        /// choose will wait for the first of five possible input data types 
        /// to arrive and will return 
        /// </summary>
        /// <typeparam name="Payload0">
        /// The first payload type 
        /// </typeparam>
        /// <typeparam name="Payload1">
        /// The second payload type 
        /// </typeparam>
        /// <typeparam name="Payload2">
        /// The third payload type 
        /// </typeparam>
        /// <typeparam name="Payload3">
        /// The fourth payload type 
        /// </typeparam>
        /// <typeparam name="Payload4">
        /// The fifth payload type 
        /// </typeparam>
        /// <param name="data0">
        /// The first reference used to capture data.
        /// </param>
        /// <param name="data1">
        /// The second reference used to capture data.
        /// </param>
        /// <param name="data2">
        /// The third reference used to capture data.
        /// </param>
        /// <param name="data3">
        /// The fourth reference used to capture data.
        /// </param>
        /// <param name="data4">
        /// The fifth reference used to capture data.
        /// </param>
        /// <returns>
        /// The (zero-based) index of the item it picked up. In other words, 
        /// if data0 is filled in, the return value will be 0, if data1 is filled
        /// in, it will be 1, etc..
        /// </returns>
        template<typename Payload0, typename Payload1, typename Payload2, typename Payload3, typename Payload4, typename Func0, typename Func1, typename Func2, typename Func3, typename Func4>
        int choose(Payload0 &data0, Payload1 &data1, Payload2 &data2, Payload3 &data3, Payload4 &data4, Func0 &action0, Func1 &action1, Func2 &action2, Func3 &action3, Func4 &action4)
        {
            _validate();

            int idx = -1;

            const char *type_name_0 = typeid(Payload0).name();
            const char *type_name_1 = typeid(Payload1).name();
            const char *type_name_2 = typeid(Payload2).name();
            const char *type_name_3 = typeid(Payload3).name();
            const char *type_name_4 = typeid(Payload4).name();

            data_lock.lock();

            if ( msg_inbox.size() > 0 )
            {
                general_msg *msg = msg_inbox.front();

                if ( msg != nullptr && msg->Matches(type_name_0) )
                    idx = 0;
                else if ( msg != nullptr && msg->Matches(type_name_1) )
                    idx = 1;
                else if ( msg != nullptr && msg->Matches(type_name_2) )
                    idx = 2;
                else if ( msg != nullptr && msg->Matches(type_name_3) )
                    idx = 3;
                else if ( msg != nullptr && msg->Matches(type_name_4) )
                    idx = 4;

                switch (idx)
                {
                case 0:
                    {
                        msg_inbox.pop();

                        bool result = _actor_info::process_message(msg, data0);
                        delete msg;

                        data_lock.unlock();

                        if ( result ) action0(true, data0);

                        return result ? 0 : -1;
                    }
                case 1:
                    {
                        msg_inbox.pop();

                        bool result = _actor_info::process_message(msg, data1);
                        delete msg;

                        data_lock.unlock();

                        if ( result ) action1(true, data1);

                        return result ? 1 : -1;
                    }
                case 2:
                    {
                        msg_inbox.pop();

                        bool result = _actor_info::process_message(msg, data2);
                        delete msg;

                        data_lock.unlock();

                        if ( result ) action2(true, data2);

                        return result ? 2 : -1;
                    }
                case 3:
                    {
                        msg_inbox.pop();

                        bool result = _actor_info::process_message(msg, data3);
                        delete msg;

                        data_lock.unlock();

                        if ( result ) action3(true, data3);

                        return result ? 3 : -1;
                    }
                case 4:
                    {
                        msg_inbox.pop();

                        bool result = _actor_info::process_message(msg, data4);
                        delete msg;

                        data_lock.unlock();

                        if ( result ) action4(true, data4);

                        return result ? 4 : -1;
                    }
                }
            }
                   
            auto my_pid = self_pid;

            auto forwarder = CreateForwarder(true, [=,this](general_msg *msg) -> bool
            {
                // Silently consume messages if not valid
                if ( msg == nullptr || !my_pid->is_valid() ) 
                {
                    delete msg;
                    return false;
                }

                int idx = -1;

                if ( msg->Matches(type_name_0) )
                    idx = 0;
                else if ( msg->Matches(type_name_1) )
                    idx = 1;
                else if ( msg->Matches(type_name_2) )
                    idx = 2;
                else if ( msg->Matches(type_name_3) )
                    idx = 3;
                else if ( msg->Matches(type_name_4) )
                    idx = 4;

                switch (idx)
                {
                case 0:
                    Payload0 data0;

                    if ( _actor_info::process_message(msg, data0) )
                    {
                        proc_lock.lock();

                        if ( my_pid->is_valid() )
                            try
                            {
                                action0(false, data0);
                                if ( my_pid->is_valid() ) proc_lock.unlock();
                            }
                            catch(...)
                            {
                                log::post(LOG_ERROR, "Exception caught in accept callback");
                                fault(Exception);
                            }
                    }
                    break;
                case 1:
                    Payload1 data1;

                    if ( _actor_info::process_message(msg, data1) )
                    {
                        proc_lock.lock();

                        if ( my_pid->is_valid() )
                            try
                            {
                                action1(false, data1);
                                if ( my_pid->is_valid() ) proc_lock.unlock();
                            }
                            catch(...)
                            {
                                log::post(LOG_ERROR, "Exception caught in accept callback");
                                fault(Exception);
                            }
                    }
                    break;
                case 2:
                    Payload2 data2;

                    if ( _actor_info::process_message(msg, data2) )
                    {
                        proc_lock.lock();

                        if ( my_pid->is_valid() )
                            try
                            {
                                action2(false, data2);
                                if ( my_pid->is_valid() ) proc_lock.unlock();
                            }
                            catch(...)
                            {
                                log::post(LOG_ERROR, "Exception caught in accept callback");
                                fault(Exception);
                            }
                    }
                    break;
                case 3:
                    Payload3 data3;

                    if ( _actor_info::process_message(msg, data3) )
                    {
                        proc_lock.lock();

                        if ( my_pid->is_valid() )
                            try
                            {
                                action3(false, data3);
                                if ( my_pid->is_valid() ) proc_lock.unlock();
                            }
                            catch(...)
                            {
                                log::post(LOG_ERROR, "Exception caught in accept callback");
                                fault(Exception);
                            }
                    }
                    break;
                case 4:
                    Payload4 data4;

                    if ( _actor_info::process_message(msg, data4) )
                    {
                        proc_lock.lock();

                        if ( my_pid->is_valid() )
                            try
                            {
                                action4(false, data4);
                                if ( my_pid->is_valid() ) proc_lock.unlock();
                            }
                            catch(...)
                            {
                                log::post(LOG_ERROR, "Exception caught in accept callback");
                                fault(Exception);
                            }
                    }
                    break;
                }

                delete msg;
                return true;
            });

            _add_waiting_accept(type_name_0, forwarder);
            _add_waiting_accept(type_name_1, forwarder);
            _add_waiting_accept(type_name_2, forwarder);
            _add_waiting_accept(type_name_3, forwarder);
            _add_waiting_accept(type_name_4, forwarder);

            data_lock.unlock();

            return -1;
        }
    #pragma endregion

    protected:

        _DISPIMP tcp_listener *host() const { return m_host; }

        _DISPIMP actor_context get_context() const { return m_context; }

        _DISPIMP virtual bool can_migrate() const { return true; }

        _DISPIMP virtual void schedule(pplx::TaskProc, void*);

    #pragma region "private declarations"
    private:
        friend class _actor_info;
        friend class tcp_listener;
        friend class PID;

        // 
        // Private member functions
        //
        _DISPIMP void set_context(tcp_listener *host, actor_context context)
        {
            if (host) m_host = host;
            if (context) m_context = context; 
        }

        _DISPIMP actor(actor_info p_info);

        _DISPIMP friend size_t (::calculate_size)(const actor *proc);
        _DISPIMP friend bool (::serialize)(const actor *proc, actors::Serializer &serializer);
        _DISPIMP friend bool (::deserialize)(actor *proc, actors::Serializer &serializer);
        //_DISPIMP friend bool (::deserialize)(PID &pid, actor *proc, actors::Serializer &serializer);

        _DISPIMP static bool _clear_storage(const actor *proc);
        _DISPIMP static bool _recover(const msg_guid &guid, storage_block &data);

        void _validate() const { if ( !self_pid->is_valid() ) { already_faulted_exc exc = None; throw exc; } }

        _DISPIMP void _add_link(PID);

        _DISPIMP void _release_order();
        _DISPIMP bool _enforce_order();

        _DISPIMP virtual bool dispatch(dispatch_msg *);
        _DISPIMP bool _internal_dispatch(dispatch_msg *msg);

        _DISPIMP bool _run_scheduled();
        _DISPIMP bool _schedule(IMessageDispatcher, dispatch_msg *);

        _DISPIMP void _add_waiting_accept(const std::string &type, IMessageDispatcher dispatcher);
        _DISPIMP IMessageDispatcher _get_waiting_accept(const std::string &type);

        actor_info _proc_info() { return self_pid->m_actor_info; }

        //
        // Private data
        //

        std::shared_ptr<PID> self_pid;

        tcp_listener *m_host;
        actor_context m_context;

        __declspec(thread) static actor *s_current;

        bool faulted;

        volatile long ordering_flag;
        volatile long processing_flag;
        volatile long executing_flag;

        std::queue<dispatch_msg *> to_process;
        std::queue<general_msg *> msg_inbox;

        std::queue<std::pair<IMessageDispatcher, dispatch_msg *>> waiting_to_run;

        Concurrency::critical_section proc_lock;
        Concurrency::critical_section data_lock;

        PID::Group linked_processes;
        std::hash_map<std::string, IMessageDispatcher> attached_dispatchers;
        std::deque<std::pair<std::string, IMessageDispatcher>> waiting_accepts;

        storage_block checkpoint_data;

        static std::shared_ptr<ProcessStorage> s_storage;
        static std::shared_ptr<ProcessDirectory> s_directory;
#pragma endregion
    };

} // namespace actors
